package mapreduce.phase1;

import java.io.IOException;
import java.util.StringTokenizer;

import mapreduce.ColumnGrp;
import mapreduce.DataToPass;
import mapreduce.Point;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

/**
 * Localized run on points of the column to find connected components 
 */

@SuppressWarnings("deprecation")
public class Map1 extends MapReduceBase implements
	Mapper<LongWritable, Text, IntWritable, Text> {
		private final static IntWritable part = new IntWritable(1);

		public void map(LongWritable key, Text value,
				OutputCollector<IntWritable, Text> output, Reporter reporter)
		throws IOException {
			String line = value.toString();
			StringTokenizer tokenizer = new StringTokenizer(line);
			assert (tokenizer.hasMoreTokens());
			double x = Double.parseDouble(tokenizer.nextToken());
			assert (tokenizer.hasMoreTokens());
			double y = Double.parseDouble(tokenizer.nextToken());
			assert (tokenizer.hasMoreTokens());
			double w = Double.parseDouble(tokenizer.nextToken());
			if (w >= DataToPass.wmin && w <= DataToPass.wmax) {
				Point p = new Point(x, y);
				Text t = new Text(p.toString());
				part.set(ColumnGrp.getMainGroup(p));
				output.collect(part, t);
				int secondryGroup = ColumnGrp.getSecondryGroup(p);
				if(secondryGroup!=-1) {
					part.set(secondryGroup);
					output.collect(part, t);
				}
			}
		}
	}
